// @Author EthanScriptOn
// @Desc
package rabbitmq_operator

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_param_collector"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
)

type RabbitmqProduceOperator struct {
	*identity.IdentityInformationCarrier
}

func (r *RabbitmqProduceOperator) Execute(ctx *actuator.SrfSwitchBridgeActuatorContext, configIdentifier interface{}, structureParam interface{}) (interface{}, error) {
	rabbitmqProducerParams, ok := structureParam.(*extend_param_collector.RabbitmqProducerParams)
	if !ok {
		return nil, fmt.Errorf("params is not RabbitmqProducerParams")
	}
	if !rabbitmqProducerParams.IsValid() {
		return nil, fmt.Errorf("RabbitmqProducerParams is not valid")
	}

	rabbitmqProducer, err := connect.GetRabbitmqProducer()
	if err != nil {
		return nil, fmt.Errorf("get rabbitmq producer failed, err: %v", err)
	}

	responses := make([]*extend_config.RabbitmqOperatorResponse, 0)
	for _, param := range rabbitmqProducerParams.Params {
		err := rabbitmqProducer.Publish(
			param.Exchange,
			param.Routingkey,
			param.Mandatory,
			param.Immediate,
			*param.Msg,
		)
		resp := new(extend_config.RabbitmqOperatorResponse)
		resp.Exchange = param.Exchange
		resp.Routingkey = param.Routingkey
		resp.OriginMsg = param.Msg
		resp.Err = err
		responses = append(responses, resp)
	}
	return responses, nil
}
